#include "transContentPortListen.hpp"

///////////////////////////////////////////////////////////////////////
/// accept connect request
///		the accept just accept connect request when the acceptable count not reached the max limit
///		after all accept sequence finished, the accept finished, then return value to users.
///
/// return:
///		E_TRANC_ACCEPT_FULL: accpt full
///		E_TRANC_TIMEOUT: wait timeout
///		E_TRANC_NONBLOCK: the caller use sync style
///		vid: the new stream record index
////////////////////////////////////////////////////////////////////////
vroute_t TransContentPortListen::accept(){
	debug_enter();
	timespec ts;
	
	if(acceptState != 0){
		logd_error("the accept port is accepting, can not call accept again");
		return E_TRANC_InsStateFailed;
	}

	///check left count
	if(leftAllowedCount <= 0){
		logd_error("accept count reached the max count  limits");
		return E_TRANC_ACCEPT_FULL;
	}

	acceptState = 1; //accepting

	///wait or return;
	if(callback != NULL){
		logd_info("use sync style, the caller not block, now return nonblock");
		return E_TRANC_NONBLOCK;
	}
Start:
	logd_info("use block style, the caller will block, begin to wait");
	ts = TransContentAssis::TimeTp2Ts(timeout);
	pthread_mutex_lock(&waitMutex);
	if(pthread_cond_timedwait(&waitCond, &waitMutex, &ts) == ETIMEDOUT){
		logd_info("accept, wait timed out: %d", timeout);
		acceptState = 0; //timed out set the state to: not start
		acceptResult = 3;
		pthread_mutex_unlock(&waitMutex);
		return E_TRANC_TIMEOUT;
	}
	/// connect request come
	pthread_mutex_unlock(&waitMutex);
	/// now set result to 2 not 0(failed), 1(sucessfully)
	if(acceptResult == 2){
		acceptResult = 0;
		logd_error("accept next step may be create new stream port fialed, so this accept not really, accept again");
		goto Start;
	}
	//accept sucessfully
	--leftAllowedCount;
	acceptState = 0; // accept finished, can accept again set to 0
	return vid;
}


int TransContentPortListen::response(char* content, int index){
	int subMessage = 0;
	int mainCode = content[0];
	ConnectPeer remotePeer;

	if(mainCode != message_type_accept-connect){
		logd_error("response, the main code not supported in listen instance");
		return RESPONSE_RESULT_POST_CONITNUE;
	}

	subMessage = content[1];
	switch(subMessage){
		case message_type_accept-connect_sub_connect:
			logd_info("response accept connect start message");
			acceptConnect(content, &remotePeer);
			break;
		default:
			logd_error("accept not support sub message:%d", subMessage);
	}
	logd_info("start to release the buffer in memory pool : %d", index);
	TransContentMemory::getInstance()->releaseBuffer(index);
	// accept state = 2, means the accept is create a new stream record, so the accept can not accept a port at this time
	if(acceptState == 2){
		logd_error("a connect request is doing, this new packet will ignore");
		acceptACK(&remotePeer, ACCEPT_PORT_REFUSED);
		return RESPONSE_RESULT_NOACTION;
	}
	acceptState = 2;
	mPeer = remotePeer;
	return RESPONSE_RESULT_CREATE_STREAM;
}

void TransContentPortListen::acceptConnect(char* content, ConnectPeer* remotePeer){
	debug_enter();
	// begin extract data info
	if(remotePeer != NULL){
		remotePeer->acceptPort = localPort;
		remotePeer->remotePort = TransContentAssis::INTN2L(*(int*(content+6)));
		remotePeer->remoteAddr = TransContentAssis::NodeAddrN2L(*((NodeAddr*)(content+10)));
	}

	logd_info("extract data form accept request is: remotePort: %d, remoteAddr:%lld", remotePeer->remotePort, remotePeer->remoteAddr.serial_id);
	debug_leave();
}


/// port manager to notice the instance, that the stream record created finished
void TransContentPortListen::acceptFinished(int result){
	debug_enter();

	acceptResult = result;
	if(callback != NULL && acceptResult == 1){
		logd_info("accept use sync style, so if result == 1, will call the callback else do nothing");
		transContentThreadPool::getInstance()->addTask(finishRoute, this);
		return;
	}
	logd_info("accept finished, the the blocked thread go on");
	pthread_mutex_lock(&waitMutex);
	pthread_cond_signal(&waitCond);
	pthread_mutex_unlock(&waitMutex);
	logd_info("let the accept block to run again");
	return;
}

void*  TransContentPortListen::finishRoute(void* p){
	debug_enter();
	(TransContentPortListen*(p))->finishedCallback();
	return NULL;
}

void  TransContentPortListen::finishedCallback(){
	debug_enter();
	--leftAllowedCount;
	acceptState = 0;
	callback(streamVid);
	debug_leave();
}

void TransContentPortListen::acceptACK(ConnectPeer* remotePeer, int result){
	char content[24]={0};
	debug_enter();
	logd_info("begin to construct ack packet");
	*(short*(content)) = (short)19; //total length exclude total length
	content[2] = (char)message_type_accept-connect;
	*(NodeAddr*(content+3)) = remotePeer->remoteAddr;
	content[11] = (char)SEND_FLAG_USE_PATH_OPTION;
	*(short*(content+12)) = TransContentAssis::SHORTL2N(7);
	content[14] = (char)message_type_accept-connect;
	content[15] = (char)message_type_accept-connect_sub_connectAck;
	*(int*(content+16)) = TransContentAssis::INTL2N(remotePeer->remotePort);
	content[20] = (char)result;
	logd_info("construct ack packet finished, start send to stack");
	TransContentDispatcher::getInstance()->sendDataDataToStack(content, 21);
	logd_info("send packet finished");
	debug_leave();
}
